package com.atguigu.tingshu.common.kafka.service;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.common.kafka.entity.KafkaMessageData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * Kafka 消息发送服务
 */
@Slf4j
@Service
public class KafkaService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 发送普通消息
     *
     * @param topic   主题
     * @param message 消息内容
     * @return 是否发送成功
     */
    public boolean sendMessage(String topic, Object message) {
        return sendMessage(topic, null, message);
    }

    /**
     * 发送普通消息（带分区键）
     *
     * @param topic        主题
     * @param partitionKey 分区键（用于控制消息发送到哪个分区）
     * @param message      消息内容
     * @return 是否发送成功
     */
    public boolean sendMessage(String topic, String partitionKey, Object message) {
        try {
            // 1. 创建消息数据对象
            KafkaMessageData messageData = new KafkaMessageData();
            String messageId = "kafka:" + UUID.randomUUID().toString().replaceAll("-", "");
            messageData.setMessageId(messageId);
            messageData.setTopic(topic);
            messageData.setMessage(message);
            messageData.setPartitionKey(partitionKey);
            messageData.setTimestamp(System.currentTimeMillis());

            // 2. 将消息数据存入 Redis（用于重试和幂等性校验）
            redisTemplate.opsForValue().set(messageId, JSON.toJSONString(messageData), 10, TimeUnit.MINUTES);

            // 3. 发送消息到 Kafka
            String messageJson = JSON.toJSONString(message);
            CompletableFuture<SendResult<String, String>> future;

            if (partitionKey != null && !partitionKey.isEmpty()) {
                // 使用分区键发送
                future = kafkaTemplate.send(topic, partitionKey, messageJson);
            } else {
                // 不指定分区键
                future = kafkaTemplate.send(topic, messageJson);
            }

            // 4. 异步处理发送结果
            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    // 发送成功
                    log.info("消息发送成功 - Topic: {}, Partition: {}, Offset: {}, MessageId: {}",
                            result.getRecordMetadata().topic(),
                            result.getRecordMetadata().partition(),
                            result.getRecordMetadata().offset(),
                            messageId);
                } else {
                    // 发送失败，进行重试
                    log.error("消息发送失败 - Topic: {}, MessageId: {}, Error: {}",
                            topic, messageId, ex.getMessage());
                    retrySendMessage(messageData);
                }
            });

            return true;
        } catch (Exception e) {
            log.error("发送 Kafka 消息异常 - Topic: {}, Error: {}", topic, e.getMessage(), e);
            return false;
        }
    }

    /**
     * 发送延迟消息
     * 注意：Kafka 本身不支持延迟消息，这里使用 Redis 实现简单的延迟功能
     * 生产环境建议使用专门的延迟队列方案（如 RocketMQ 或自定义实现）
     *
     * @param topic     主题
     * @param message   消息内容
     * @param delayTime 延迟时间（秒）
     * @return 是否发送成功
     */
    public boolean sendDelayMessage(String topic, Object message, int delayTime) {
        return sendDelayMessage(topic, null, message, delayTime);
    }

    /**
     * 发送延迟消息（带分区键）
     *
     * @param topic        主题
     * @param partitionKey 分区键
     * @param message      消息内容
     * @param delayTime    延迟时间（秒）
     * @return 是否发送成功
     */
    public boolean sendDelayMessage(String topic, String partitionKey, Object message, int delayTime) {
        try {
            // 1. 创建消息数据对象
            KafkaMessageData messageData = new KafkaMessageData();
            String messageId = "kafka:delay:" + UUID.randomUUID().toString().replaceAll("-", "");
            messageData.setMessageId(messageId);
            messageData.setTopic(topic);
            messageData.setMessage(message);
            messageData.setPartitionKey(partitionKey);
            messageData.setDelay(true);
            messageData.setDelayTime(delayTime);
            messageData.setTimestamp(System.currentTimeMillis());

            // 2. 将延迟消息存入 Redis，设置过期时间为延迟时间
            // 注意：这里只是简单实现，生产环境建议使用专门的延迟队列
            String delayKey = "kafka:delay:" + messageId;
            redisTemplate.opsForValue().set(delayKey, JSON.toJSONString(messageData), delayTime, TimeUnit.SECONDS);

            log.info("延迟消息已存储 - Topic: {}, DelayTime: {}s, MessageId: {}", topic, delayTime, messageId);
            log.warn("注意：Kafka 原生不支持延迟消息，当前使用 Redis 简单实现，生产环境请使用专门的延迟队列方案");

            // 3. 这里可以配合定时任务扫描 Redis 中的延迟消息并发送
            // 或者使用 Redis 的 keyspace notifications 功能

            return true;
        } catch (Exception e) {
            log.error("发送延迟消息异常 - Topic: {}, Error: {}", topic, e.getMessage(), e);
            return false;
        }
    }

    /**
     * 重试发送消息
     *
     * @param messageData 消息数据
     */
    private void retrySendMessage(KafkaMessageData messageData) {
        try {
            // 获取重试次数
            int retryCount = messageData.getRetryCount();
            if (retryCount >= 3) {
                log.error("消息发送失败，超过最大重试次数 - Topic: {}, MessageId: {}",
                        messageData.getTopic(), messageData.getMessageId());
                // TODO: 可以将失败消息存入数据库，人工处理
                return;
            }

            // 增加重试次数
            retryCount++;
            messageData.setRetryCount(retryCount);

            // 更新 Redis 中的消息数据
            redisTemplate.opsForValue().set(
                    messageData.getMessageId(),
                    JSON.toJSONString(messageData),
                    10,
                    TimeUnit.MINUTES
            );

            // 创建 final 变量用于 Lambda 表达式
            final int currentRetryCount = retryCount;

            log.info("开始重试发送消息 - Topic: {}, MessageId: {}, RetryCount: {}",
                    messageData.getTopic(), messageData.getMessageId(), currentRetryCount);

            // 重新发送消息
            String messageJson = JSON.toJSONString(messageData.getMessage());
            CompletableFuture<SendResult<String, String>> future;

            if (messageData.getPartitionKey() != null && !messageData.getPartitionKey().isEmpty()) {
                future = kafkaTemplate.send(messageData.getTopic(), messageData.getPartitionKey(), messageJson);
            } else {
                future = kafkaTemplate.send(messageData.getTopic(), messageJson);
            }

            // 处理重试结果
            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    log.info("消息重试发送成功 - Topic: {}, MessageId: {}, RetryCount: {}",
                            messageData.getTopic(), messageData.getMessageId(), currentRetryCount);
                } else {
                    log.error("消息重试发送失败 - Topic: {}, MessageId: {}, RetryCount: {}, Error: {}",
                            messageData.getTopic(), messageData.getMessageId(), currentRetryCount, ex.getMessage());
                    // 继续重试
                    retrySendMessage(messageData);
                }
            });

        } catch (Exception e) {
            log.error("重试发送消息异常 - MessageId: {}, Error: {}",
                    messageData.getMessageId(), e.getMessage(), e);
        }
    }

}

